package org.zcl.rocketmq.controller;

import com.alibaba.fastjson.JSONObject;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
import org.zcl.rocketmq.service.ProducerService;

import java.util.List;
import java.util.Set;

/**
 * @Package org.zcl.controller
 * @PROJECT_NAME: question
 * @DESCRIPTION:
 * @USER: zhuchenglin
 * @DATE: 2021/12/13 10:34
 */
@Api(value = "rocketMq发送消息", tags = "rocketMq发送消息")
@RestController
@Slf4j
public class TestController {

    @Autowired
    private ProducerService producerService;

    @PostMapping("push")
    @ApiOperation(value = "发消息", httpMethod = "POST")
    public void push() {
        producerService.syncGeneralMQMessageSend();
    }

    @ApiOperation(value = "测试消费者拉取消息", notes = "测试消费者拉取消息")
    @GetMapping("/test/getmessage")
    public void testConsumer() throws Exception {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("user_consumer_group");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setInstanceName("user_consumer_instance");
        consumer.start();
        System.out.println("消费者创建完成");

        Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues("user-topic");
        System.out.println("拉取到的队列数量" + messageQueues.size());

        for (MessageQueue queue : messageQueues) {
            System.out.println("遍历队列queue" + queue);

            long offset = consumer.fetchConsumeOffset(queue, true);
            System.out.println("consumer from the queue:" + queue + ":" + offset);

            for (int i = 0; i < 10; i++) {
                // 在队列中拉取不到消息会一直阻塞等待着，直到能拉取到消息
//                PullResult pullResult = consumer.pullBlockIfNotFound(queue, null, consumer.fetchConsumeOffset(queue, false), 1);

                // 在队列中拉取不到消息就结束
                PullResult pullResult = consumer.pull(queue, "", consumer.fetchConsumeOffset(queue, false), 1);

                consumer.updateConsumeOffset(queue, pullResult.getNextBeginOffset());
                switch (pullResult.getPullStatus()) {
                    case FOUND:
                        List<MessageExt> messageExtList = pullResult.getMsgFoundList();
                        for (MessageExt m : messageExtList) {
                            System.out.println("拉取到数据====" + JSONObject.toJSONString(m));
                        }
                        break;
                    case NO_MATCHED_MSG:
                        break;
                    case NO_NEW_MSG:
                        break;
                    case OFFSET_ILLEGAL:
                        break;
                    default:
                        break;
                }
            }
        }

        System.out.println("关闭消费者");
        consumer.shutdown();
    }
}
